#pragma once
#include <string>
#include <functional>
#include "base/ringbuffer.h"
#include <vector>

namespace galois::mid {

void SubProcessInit();
int  GetChannelID(std::string&);
void pub(int ch,int len,std::function<void(void*)>);
int sub(int ch,uint64_t index,std::function<void(void*,int)>);

struct ShmPubChannel {
  std::string topic;
  int channel = -1;
  std::vector<std::string> args_from_framework;
};
struct ShmPubMsg {
  std::function<void(void *)> f;

  int len;
};

int shm_puber(ShmPubChannel *channel,  ShmPubMsg *msg);
template<class T>
int shm_proto_puber(ShmPubChannel *ch,T* val){
  if(ch->channel < 0){
    //SubProcessInit();
    ch->topic = ch->args_from_framework[0];
    ch->channel = GetChannelID(ch->topic);    
  }
  auto len = val->ByteSizeLong();
  pub(ch->channel,len, [val,len](void *addr){
    val->SerializeToArray(addr,len);
  });
  return 1;
}
template <class T, int N>
struct ShmSubData {
  std::string topic;
  std::vector<std::string> args_from_framework;
  int channel = -1;
  base::RingPtr1W1R<T, N, 0> buffer;
  uint64_t index = 0;
};

template <class T, int N>
int shm_suber(ShmSubData<T, N> *ch) {
  if(ch->channel < 0){
   // SubProcessInit();
    ch->topic = ch->args_from_framework[0];
    ch->channel = GetChannelID(ch->topic);    
  }
  int ret = 0;
  int re = 0;
  do {
    T *val = nullptr;
    re = sub(ch->channel, ch->index, [&val, ch](void *addr, int len) {
      val = ch->buffer.Get();
      val->ParseFromArray(addr, len);
    });
    if (re == 1) {
      ch->index++;
      ch->buffer.push(val);
      ret++;
    } else {
      if (val) ch->buffer.revert(val);
      ch->index -= re;
    }
  } while (re != 0);
  return ret;
}
template<class T,int N>
int shm_sub_pop(T* out,ShmSubData<T, N> *ch){
  return ch->buffer.pop([out](const T *ms) {
    *out  = *ms;
  })?1:0; 
}

struct shm_config{
  uint32_t size = 0;
  uint64_t hz =0;
  uint64_t total = 0;
  std::vector<std::string> args_from_framework;
};
int shm_server(shm_config* config);

} // namespace galois::mid